package flinkSql.source;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;
import org.junit.Test;

public class Flink_SQL_FileSource {
    @Test
    public void test() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.connect(new FileSystem().path("input/sensorreading.txt"))
                .withFormat(new Csv())
                .withSchema(new Schema()
                    .field("id",DataTypes.STRING())
                    .field("ts",DataTypes.BIGINT())
                    .field("temp",DataTypes.DOUBLE()))
                .createTemporaryTable("fileSensor");
        Table table = tableEnv.sqlQuery("select id,count(temp) from fileSensor group by id");
        tableEnv.toRetractStream(table,Row.class).print("Sql");
        env.execute();
    }
}
